package jwebtest.rpc.rpc.scoket;

import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.SocketTimeoutException;
import java.util.concurrent.atomic.AtomicInteger;

import jwebtest.rpc.rpc.IRpcFramework;
import jwebtest.rpc.zk.ZNode;

public class ScoketRPC extends IRpcFramework {

	public ScoketRPC(int port, int timeout) throws IOException {
		super(port, timeout);
		startScoket();
	}

	public ScoketRPC(String postUrl, int port, int timeout) {
		super(postUrl, port, timeout);
	}

	public ScoketRPC(int timeout) {
		super(timeout);
	}

	private AtomicInteger postCount = new AtomicInteger(0);

	// ==============================
	private void startScoket() throws IOException {
		System.out.println("Export RPC server on port " + port);
		@SuppressWarnings("resource")
		final ServerSocket scoketServer = new ServerSocket(port);
		new Thread(new Runnable() {
			@Override
			public void run() {
				try {
					System.out.println("The port [" + port + "] Service is starting !  [ok]");
					while (true) {
						try {
							final Socket socket = scoketServer.accept();// 瓶颈所在
							socket.setSoTimeout(timeout);
							new Thread(new Runnable() {
								@Override
								public void run() {
									try {
										System.out.println("The host [" + socket.getInetAddress().getHostAddress() + "] is client");
										ObjectInputStream input = new ObjectInputStream(socket.getInputStream());
										try {
											// ==
											String methodName = input.readUTF();
											String serverClass = (String) input.readObject();
											Class<?>[] parameterTypes = (Class<?>[]) input.readObject();
											Object[] arguments = (Object[]) input.readObject();
											//
											Object result = null;
											ObjectOutputStream output = new ObjectOutputStream(socket.getOutputStream());
											Object serverImpl = implServer.get(serverClass);
											try {
												if (serverImpl == null) {
													throw new NullPointerException("Don't find provider[" + serverClass
															+ "],Please check if the providers have been started and registered.");
												} else {
													Method method = serverImpl.getClass().getMethod(methodName, parameterTypes);
													result = method.invoke(serverImpl, arguments);
													output.writeObject(result);
												}
											} catch (Throwable t) {
												output.writeObject(t);
											} finally {
												output.close();
											}
										} finally {
											System.out.println("The host [" + socket.getInetAddress().getHostAddress() + "] is over,No:"
													+ postCount.addAndGet(1));
											input.close();
										}
									} catch (Exception e) {
										try {
											socket.close();
										} catch (IOException e1) {
											e1.printStackTrace();
										}
										e.printStackTrace();
									}
								}
							}).start();
						} catch (Exception e) {
							e.printStackTrace();
						}
					}
				} catch (Exception e) {
					e.printStackTrace();
				}
			}
		}).start();
	}

	@Override
	public boolean addExport(Object server, int priority, int weight, String version) {
		if (server == null)
			throw new IllegalArgumentException("server instance == null");
		Class<?> interfaceName = server.getClass().getInterfaces()[0];
		implServer.put(interfaceName.getName(), server);
		if (zkApi != null) {
			ZNode zn = new ZNode(ZNodePath, interfaceName.getName(), localHost, port, priority, weight, version, null, null);
			createNode(zn, 1);
		}
		System.out.println("serverImpl[" + server.getClass().getName() + "]  export succ!");
		return true;
	}

	@Override
	public boolean addExport(Object server, int priority, int weight) {
		return addExport(server, priority, weight, defVersion);
	}

	@Override
	public boolean addExport(Object server) {
		return addExport(server, 1, 100, defVersion);
	}

	@SuppressWarnings("unchecked")
	@Override
	public <T> T reference(final Class<T> _interface, String version) {
		if (_interface == null)
			throw new IllegalArgumentException("Interface class == null");
		if (!_interface.isInterface())
			throw new IllegalArgumentException("The " + _interface.getName() + " must be interface class!");
		final Boolean[] isLine = new Boolean[] { true };
		if (zkApi != null) {
			isLine[0] = false;
			ZNode zn = new ZNode(ZNodePath, _interface.getName(), localHost, port, 0, 0, version, null, null);
			createNode(zn, 2);
		}
		return (T) Proxy.newProxyInstance(_interface.getClassLoader(), new Class<?>[] { _interface }, new InvocationHandler() {
			public Object invoke(Object proxy, Method method, Object[] arguments) throws Throwable {
				Socket socket = null;
				if (isLine[0]) {
					socket = new Socket(host, port);
					System.out.println("Get remote service " + _interface.getName() + " from server " + host + ":" + port);
				} else {
					ZNode zn = getDomainServer(_interface.getName(), 3);
					socket = new Socket(zn.getHost(), zn.getPort());
					System.out.println("Get remote service " + _interface.getName() + " from server " + zn.getHost() + ":" + zn.getPort());
				}
				socket.setSoTimeout(timeout);
				try {
					ObjectOutputStream output = new ObjectOutputStream(socket.getOutputStream());
					try {
						output.writeUTF(method.getName());
						output.writeObject(_interface.getName());
						output.writeObject(method.getParameterTypes());
						output.writeObject(arguments);
						ObjectInputStream input = new ObjectInputStream(socket.getInputStream());
						try {
							Object result = input.readObject();
							if (result instanceof Throwable) {
								throw (Throwable) result;
							}
							return result;
						} catch (SocketTimeoutException ste) {
							System.err.println("资源请求超时[" + timeout + "s]:" + _interface.getName() + method.getName());
							throw ste;
						} finally {
							input.close();
						}
					} finally {
						output.close();
					}
				} finally {
					socket.close();
				}
			}
		});
	}

	@Override
	public <T> T reference(final Class<T> _interface) {
		return reference(_interface, defVersion);
	}

}
